package top.frsf.mqtt.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.annotation.Configuration;
import top.frsf.mqtt.properties.ServerProperties;
import top.frsf.mqtt.initializer.ChannelInitializer;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

import static top.frsf.mqtt.constants.ThreadNameConstant.*;

@Slf4j
@Configuration
public class MqttServer implements SmartLifecycle {
    /**
     * MQTT 服务器配置
     */
    @Resource
    private ServerProperties properties;
    /**
     * MQTT ChannelInitializer
     */
    @Resource
    private ChannelInitializer serverInitializer;

    /**
     * Boss 线程组
     */
    private EventLoopGroup bossGroup;
    /**
     * Worker 线程组
     */
    private EventLoopGroup workerGroup;

    /**
     * MQTT 服务器运行线程池
     */
    private ExecutorService serverExecutor;
    /**
     * Server启动状态
     */
    private final AtomicBoolean running = new AtomicBoolean(false);


    @Override
    public void start() {
        boolean state = running.get();
        // 启动状态直接返回
        if (state) {
            return;
        }
        // 非启动状态，创建单线程pool
        DefaultThreadFactory factory = new DefaultThreadFactory(MAIN_THREAD_NAME);
        serverExecutor = Executors.newSingleThreadExecutor(factory);
        serverExecutor.submit(this::doStart);
    }

    @Override
    public void stop() {
        if (running.compareAndSet(true, false)) {
            // 先关闭线程池
            if (serverExecutor != null) {
                serverExecutor.shutdownNow();
            }
            // 关闭netty线程组
            if (bossGroup != null) {
                bossGroup.shutdownGracefully();
            }
            if (workerGroup != null) {
                workerGroup.shutdownGracefully();
            }
            log.info("MQTT Server stopped");
        }
    }

    @Override
    public boolean isRunning() {
        return running.get();
    }

    public void doStart() {
        bossGroup = new NioEventLoopGroup(properties.getBossThreads(), new DefaultThreadFactory(BOSS_THREAD_NAME));
        workerGroup = new NioEventLoopGroup(properties.getWorkerThreads(), new DefaultThreadFactory(WORKER_THREAD_NAME));

        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, properties.getBacklog())
                    .childOption(ChannelOption.SO_KEEPALIVE, properties.isKeepAlive())
                    .childHandler(serverInitializer);

            ChannelFuture channelFuture = bootstrap.bind(properties.getPort()).addListener(new SuccessFutureListener());
            channelFuture.channel().closeFuture().addListener((ChannelFutureListener) future -> stop());
        } catch (Exception e) {
            log.error("MQTT Server error", e);
        }
    }


    /**
     * Bootstrap启动监听器
     */
    private class SuccessFutureListener implements GenericFutureListener<ChannelFuture> {
        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {
            if (channelFuture.isSuccess()) {
                running.set(true);
                successBanner();
            } else {
                log.error("Failed to bind MQTT Server on port {}", properties.getPort(), channelFuture.cause());
            }
        }
    }

    /**
     * 服务器启动成功后打印启动信息
     */
    private void successBanner() {
        String banner = "\n" +
                "* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\n" +
                "*                                                                       *\n" +
                "*              M I N I - M Q T T   B R O K E R   [ RUNNING ]            *\n" +
                "*                                                                       *\n" +
                "* * * * * * * * * * * * 版本：v1.0.0 | 端口：{} * * * * * * * * * * * * *\n";
        log.info(banner, properties.getPort());
        log.info("MQTT Server started on port {}", properties.getPort());
    }
}
